package com.shujia.mllib

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg.{SparseVector, Vectors}
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo7proImage {
  def main(args: Array[String]): Unit = {

    /**
      * 模型使用
      *
      */

    val spark: SparkSession = SparkSession.builder()
      .appName("kmeans")
      .master("local[8]")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    //1、加载模型
    val model: LogisticRegressionModel = LogisticRegressionModel.load("spark/data/imagemodel")


    //2 读取图片
    val imageData: DataFrame = spark.read
      .format("image")
      .load("spark/data/27550.jpg")

    //取出文件名数据
    val data: DataFrame = imageData.select($"image.origin" as "name", $"image.data" as "data")


    val nameandfeatures: DataFrame = data
      //将DF转换成DataSet ,才能使用map函数
      .as[(String, Array[Byte])]
      .map(kv => {
        //数据
        val value: Array[Byte] = kv._2

        //将每一个像素的转换成Double
        //将数据转换成0或者1
        val idata: Array[Double] = value
          .map(_.toDouble)
          .map(p => {
            if (p < 0) {
              1.0
            } else {
              0.0
            }
          })

        println(idata.length)

        //将数据转换成稀疏向量
        val sparse: SparseVector = Vectors.dense(idata).toSparse

        (kv._1, sparse)
      }).toDF("name", "features")


    //通过模型识别图片
    val dataFrame: DataFrame = model.transform(nameandfeatures)

    dataFrame.show()


  }

}
